package com.cancer.attend.task;

import com.cancer.attend.entity.AttendReport;
import com.cancer.attend.utils.DateUtility;
import com.cancer.attend.utils.WebUtil;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.*;

import java.util.*;

/**
 * Created by cuiwx on 2016/12/14.
 */
public class AttendReportTask {
    private static final String DB_USERNAME = "hqdb";
    private static final String DB_PWD = "hongquan";
    private static final String DB_CONNECTION_URL = "jdbc:oracle:thin:@123.56.24.226:1521:gps";
    private static final String DB_DRIVER = "oracle.jdbc.driver.OracleDriver";

    private static final String HCDB_USERNAME = "root";
    private static final String HCDB_PWD = "hecom123";
    private static final String HCDB_CONN_URL = "jdbc:mysql://101.201.239.240:3306/hecom_bi";

    public static void main(String[] args) throws Exception {
        JavaSparkContext sc = new JavaSparkContext(new SparkConf()
                .setAppName("AttendReportTask").setMaster("local[*]"));
        sc.setLogLevel("WARN");
        SQLContext sqlContext = new SQLContext(sc);

        java.sql.Date now = new java.sql.Date(System.currentTimeMillis()); //当前时间

        Map<String, String> options = new HashMap<String, String>();
        options.put("url", DB_CONNECTION_URL);
        options.put("driver", DB_DRIVER);
        options.put("user", DB_USERNAME);
        options.put("password", DB_PWD);

        DataFrame clockingDF;
        DataFrame orgDF;
        DataFrame attendDF;

        options.put("dbtable", "v30_bd_clocking_in");
        clockingDF = sqlContext.read().format("jdbc").options(options).load();

        options.put("dbtable", "V30_MD_ORGANIZATION");
        orgDF = sqlContext.read().format("jdbc").options(options).load();

        options.put("dbtable", "M60_ENT_ATTENDANCE_WHITE");
        attendDF = sqlContext.read().format("jdbc").options(options).load();

        attendDF = attendDF.select(attendDF.col("ENT_CODE"),attendDF.col("EMPLOYEE_CODE"),
                attendDF.col("START_TIME"),attendDF.col("END_TIME"),attendDF.col("STATUS"));

        attendDF.registerTempTable("ATTENDANCE_WHITE");
        clockingDF.registerTempTable("CLOCKING_IN");
        orgDF.registerTempTable("ORGANIZATION");

        StringBuilder coSql = new StringBuilder();
        coSql.append(" select c.ID,c.EMPLOYEE_CODE,c.SIGN_IN_TIME,c.SIGN_OUT_TIME, ");
        coSql.append(" c.SIGN_IN_LATITUDE,c.SING_IN_LONGITUDE,c.SIGN_IN_PLACE, ");
        coSql.append(" c.SIGN_OUT_LATITUDE,c.SING_OUT_LONGITUDE,c.SIGN_OUT_PLACE, ");
        coSql.append(" o.ENT_CODE, o.START_WORK_TIME, o.END_WORK_TIME, o.WORK_DAYS ");
        coSql.append(" from CLOCKING_IN c ");
        coSql.append(" join ORGANIZATION o on o.CODE = c.EMPLOYEE_CODE ");
        coSql.append(" where o.STATUS = 0 and o.ISEMPLOYEE = '1' and o.PARENT_CODE is not null ");

        DataFrame clockingOrgDF = sqlContext.sql(coSql.toString());

        //白名单人员CODE集合
        List<String> attendEmCodeList = attendDF.toJavaRDD().map(new Function<Row, String>() {
            @Override
            public String call(Row row) {
                return row.getString(1);
            }
        }).collect();
        Set<String> whiteCodeSet = new HashSet<String>();
        whiteCodeSet.addAll(attendEmCodeList);

        JavaRDD<AttendReport> attendReportRDD = clockingOrgDF.toJavaRDD().map(new Function<Row, AttendReport>() {
            @Override
            public AttendReport call(Row row) {
                AttendReport attendReport = new AttendReport();
                attendReport = setAttendReportValue(attendReport, row, now, whiteCodeSet);
//                attendReport.setId(new Random().nextLong());//只用于持久化到数据库
                return attendReport;
            }
        });

        //RDD-->DataFrame
        DataFrame attendReportDF = sqlContext.createDataFrame(attendReportRDD, AttendReport.class);
        attendReportDF.show();

        //Save到HDFS或本地文件
       String reportHDFS = "/hecom/ods/oms3/attend-report.parquet";
       attendReportDF.write().parquet(reportHDFS);
        // String reportFile = "file:////Users/admin/Documents/Data/attend-report.json";
        // attendReportDF.write().mode(SaveMode.Overwrite).json(reportFile);

        //Save到数据库
//        Properties properties = new Properties();
//        properties.put("user", HCDB_USERNAME);
//        properties.put("password", HCDB_PWD);
//        attendReportDF.write().mode(SaveMode.Overwrite)
//                .jdbc(HCDB_CONN_URL,"m60_attend_report", properties);

        sc.stop();
    }

    private static AttendReport setAttendReportValue(
            AttendReport attendReport, Row row,
            java.sql.Date reportDay, Set<String> whiteCodeSet){
        Long clockingRecordId = WebUtil.objectToLong(row.get(0)); //考勤记录ID
        String employeeCode = row.get(1).toString(); //人员CODE
        Long signInTime = WebUtil.objectToLong(row.get(2)); //签到时间，时间戳
        Long signOutTime = WebUtil.objectToLong(row.get(3)); //签退时间，时间戳
        String signinlatitude = WebUtil.objectToString(row.get(4)); //签到纬度
        String singinlongitude = WebUtil.objectToString(row.get(5)); //签到经度
        String signoutlatitude = WebUtil.objectToString(row.get(6)); //签退纬度
        String singoutlongitude = WebUtil.objectToString(row.get(7)); //签退经度
        String signinplace = WebUtil.objectToString(row.get(8)); //签到地址
        String signoutplace = WebUtil.objectToString(row.get(9)); //签退地址
        String entCode = row.get(10).toString(); //企业CODE
        String startWorkTime = row.get(11).toString(); //上班时间，格式 08:00
        String endWorkTime = row.get(12).toString(); //下班时间，格式 18:00
        String workDays = row.get(13).toString(); //工作日

        attendReport.setClockingRecordId(clockingRecordId);
        attendReport.setSignInTime(signInTime);
        attendReport.setSignOutTime(signOutTime);
        attendReport.setSignInLatitude(signinlatitude);
        attendReport.setSignInLongitude(singinlongitude);
        attendReport.setSignInPlace(signoutlatitude);
        attendReport.setSignOutLatitude(singoutlongitude);
        attendReport.setSignOutLongitude(signinplace);
        attendReport.setSignOutPlace(signoutplace);
        attendReport.setEntCode(entCode);
        attendReport.setEmployeeCode(employeeCode);
        attendReport.setStartWorkTime(startWorkTime);
        attendReport.setEndWorkTime(endWorkTime);
        attendReport.setWorkDays(workDays);

        //判断是否是休息日，休息日则算正常
        if(!DateUtility.isWorkDay(reportDay, workDays)) {
            attendReport.setExtraType(1); //休息日
            attendReport.setAttendType(AttendReport.TYPE_NORMAL); //算正常
        }
        //是否属于白名单
        if(whiteCodeSet.contains(employeeCode)) {
            attendReport.setExtraType(2); //白名单
            attendReport.setAttendType(AttendReport.TYPE_NORMAL); //算正常
        }
        //计算考勤情况
        Date startWorkDate = DateUtility.hhmmssToDate(reportDay, startWorkTime +  ":00"); //当天开始上班的准确时间
        Date endWorkDate = DateUtility.hhmmssToDate(reportDay, endWorkTime + ":00"); //当前下班的准确时间
        if(startWorkDate == null) {
            startWorkDate = new Date(DateUtility.getStartTimeOfDay(reportDay));
        }
        if(endWorkDate == null) {
            endWorkDate = new Date(DateUtility.getEndTimeOfDay(reportDay));
        }
        long startWorkTimeLong = startWorkDate.getTime();
        long endWorkTimeLong = endWorkDate.getTime();

        //默认
        //attendReport.setAttendType(M60AttendReport.TYPE_NORMAL); //正常
        //未签到 且 未签退
        if(signInTime < 1 && signOutTime < 1) {
            attendReport.setAttendType(AttendReport.TYPE_ABSENT); //无考勤记录
        }
        else if(signInTime < 1) { //未签到（算作迟到）
            attendReport.setAttendType(AttendReport.TYPE_LATE); //迟到
            //是否有早退
            if(signOutTime < endWorkTimeLong) {
                attendReport.setAttendType(AttendReport.TYPE_LATE_EARLY); //迟到+早退（至少有一个考勤记录）
            }
        }
        else if(signOutTime < 1) { //未签退（算作早退）
            attendReport.setAttendType(AttendReport.TYPE_EARLY); //早退
            //是否有迟到
            if(signInTime > startWorkTimeLong) {
                attendReport.setAttendType(AttendReport.TYPE_LATE_EARLY); //迟到+早退（至少有一个考勤记录）
            }
        }
        else { //签到、签退记录都有
            if(signInTime > startWorkTimeLong && signOutTime < endWorkTimeLong) {
                //迟到 且 早退
                attendReport.setAttendType(AttendReport.TYPE_LATE_EARLY);
            }
            else if(signInTime > startWorkTimeLong) {
                //只有迟到
                attendReport.setAttendType(AttendReport.TYPE_LATE);
            }
            else if(signOutTime < endWorkTimeLong) {
                //只有早退
                attendReport.setAttendType(AttendReport.TYPE_EARLY);
            }
            else {
                attendReport.setAttendType(AttendReport.TYPE_NORMAL); //正常
            }
        }

        return attendReport;
    }
}